#pragma once
#ifndef __M_SERVER_H__
#define __M_SERVER_H__
#include <vector>
#include <iostream>
#include <cassert>
#include <cstring>
#include <netinet/in.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <unistd.h>
#include <functional>
#include <sys/epoll.h>
#include <unordered_map>
#include <mutex>
#include <thread>
#include <sys/eventfd.h>
#include <memory>
#include <sys/timerfd.h>
#include <condition_variable>

#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL INF
#define LOG(level, format, ...) do\
    {\
        if (level < LOG_LEVEL)\
            break;\
        time_t t = time(nullptr);\
        struct tm *ltm = localtime(&t);\
        char tmp[32] = {0};\
        strftime(tmp, 31, "%H:%M:%S", ltm);\
        fprintf(stdout, "[%s %s:%p %s %d] " format "\n",tmp,__func__, (void *)pthread_self(), __FILE__, __LINE__, ##__VA_ARGS__);\
    }while(0)

#define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)

#define BUFFER_DEFAULT_SIZE 1024
class Buffer{
    private:
        std::vector<char> _buffer;
        uint64_t _read_idx;
        uint64_t _write_idx;
    public:
        Buffer():_read_idx(0),_write_idx(0),_buffer(BUFFER_DEFAULT_SIZE){}
        Buffer(const char*data):_read_idx(0),_write_idx(0),_buffer(BUFFER_DEFAULT_SIZE){
            //初始化，和write方法一样
            std::copy(data, data + sizeof data, WritePosition());
        }
        char *Begin() { return &(*_buffer.begin()); }
        char *ReadPosition() { return Begin() + _read_idx; }
        char *WritePosition() { return Begin() + _write_idx; }
        uint64_t TailIdleSize() { return _buffer.size() - _write_idx; }
        uint64_t ReadAbleSize() { return _write_idx - _read_idx; }
        void MoveWriteOffset(uint64_t len) { _write_idx += len; }
        void MoveReadOffset(uint64_t len) { _read_idx += len; }
        void EnableWrite(int len)
        {
            // 如果末尾空闲空间大小足够，直接返回
            if(len <= TailIdleSize()){
                return;
            }
            //末尾空闲空间不够，则判断加上起始位置的空闲空间大小是否足够, 够了就将数据移动到起始位置
            else if(len <= TailIdleSize() +_read_idx ){
                //将数据移动到起始位置
                uint64_t rsz = ReadAbleSize();//把当前数据大小先保存起来
                std::copy(ReadPosition(), ReadPosition() + rsz, Begin());
                _read_idx = 0;
                _write_idx = rsz;
            }
            else{
                //总体空间不够，则需要扩容，不移动数据，直接给写偏移之后扩容足够空间即可
                _buffer.resize(_write_idx + len);
            }
        }

        //第一种
        //数据写入缓冲区
        void Write(const void* buff,uint64_t len){
            //1.先保证有写入缓冲区的条件
            if(len==0)
                return;
            EnableWrite(len);
            // 2.写入缓冲区
            const char *b = (const char *)buff;
            std::copy(b, b + len, WritePosition());
        }
        void Read(char*buff,uint64_t len){
            //因为copy必须要指定拷贝大小，所以要保证读取大小小于可读大小
            assert(len <= ReadAbleSize());
            std::copy(ReadPosition(), ReadPosition() + len, buff);
        }
        void WriteAndPush(const void*data,uint64_t len){
            Write(data, len);
            MoveWriteOffset(len);
        }
        void ReadAndPop(char* buff,uint64_t len){
            Read(buff, len);
            MoveReadOffset(len);
        }

        //第二种
        void WriteString(const std::string& data){
            //重用第一种
            Write(data.c_str(), data.size());
        }
        std::string ReadString(uint64_t len){
            std::string data;
            //不指定大小会错误吗
            data.resize(len);
            Read(&(*data.begin()), len);
            return data;
        }
        void WriteStringAndPush(const std::string& data){
            WriteString(data);
            MoveWriteOffset(data.size());
        }
        std::string ReadStringAndPop(uint64_t len){
            std::string str = ReadString(len);
            MoveReadOffset(len);
            return str;
        }


        //第三种
        void WriteBuffer(Buffer& data){
            Write(data.ReadPosition(), data.ReadAbleSize());
        }
        void WriteBufferAndPush(Buffer& data){
            Write(data.ReadPosition(), data.ReadAbleSize());
            MoveWriteOffset(data.ReadAbleSize());
        }

        //第四种
        char* FindCRLF(){
            char* str = (char*)memchr(ReadPosition(), '\n', ReadAbleSize());
            return str;
        }
        std::string GetLine(){
            //http获取一行
            char *pos = FindCRLF();
            if(pos==nullptr)
                return "";
            // +1是为了把换行字符也取出来
            return ReadString(pos - ReadPosition() + 1);
        }
        std::string GetLineAndPop(){
            std::string str = GetLine();
            MoveReadOffset(str.size());
            return str;
        }

        //清空缓冲区
        void Clear(){
            _read_idx = 0;
            _write_idx = 0;
        }
};

#define MAX_LISTEN 1024

class Socket{
    private:
        int _sockfd;
    public:
        Socket() : _sockfd(-1){}
        Socket(int sockfd) : _sockfd(sockfd){}
        int Fd() { return _sockfd; }
        bool Create()
        {
            _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
            if(_sockfd < 0){
                DBG_LOG("Create error!");
                perror("Create error!");
                return false;
            }
            return true;
        }
        bool  Bind(const std::string&ip,uint16_t port){
            struct sockaddr_in addr;
            addr.sin_family = AF_INET;
            addr.sin_port = htons(port);
            addr.sin_addr.s_addr = inet_addr(ip.c_str());
            int ret = bind(_sockfd, (struct sockaddr*)&addr, sizeof(addr));
            if(ret < 0){
                DBG_LOG("bind error!");
                perror("bind error!");
                return false;
            }
            return true;
        }
        bool Listen(int backlog = MAX_LISTEN){

            int ret = listen(_sockfd, backlog);
            LOG(DBG, "listen ret:%d", ret);
            if(ret < 0){
                DBG_LOG("listen error!");
                perror("listen error!");
                return false;
            }
            return true;
        }
        bool Connect(const std::string& ip,uint16_t port){
            struct sockaddr_in addr;
            addr.sin_family = AF_INET;
            addr.sin_port = htons(port);
            addr.sin_addr.s_addr = inet_addr(ip.c_str());
            int ret = connect(_sockfd, (struct sockaddr*)&addr, sizeof(addr));
            if(ret < 0){
                DBG_LOG("connect error!");
                perror("connect error!");
                return false;
            }
            return true;
        }
        int Accept(){
            struct sockaddr_in addr;
            socklen_t len = sizeof(struct sockaddr);
            int newfd = accept(_sockfd, (struct sockaddr *)&addr, &len);
            if(newfd < 0){
                DBG_LOG("accept error!");
                perror("accept error!");
                return -1;
            }
            LOG(DBG, "accept newfd:%d", newfd);
            return newfd;
        }
        ssize_t Recv(char*buff,ssize_t len,int flag = 0){
            ssize_t ret = recv(_sockfd, (void *)buff, len, flag);
            if(ret <= 0){
                if(errno == EAGAIN || errno ==EINTR)
                    return 0;//表示这次接收没有接收到数据
                DBG_LOG("recv error!");
                perror("recv error!");
                return -1;
            }
            return ret;//实际接收的数据长度
        }
        ssize_t Send(const char*buff,ssize_t len,int flag = 0){
            ssize_t ret = send(_sockfd, (const void *)buff, len, flag);
            if (ret <= 0) {
                if (errno == EAGAIN || errno == EINTR) {
                    return 0;
                }
                DBG_LOG("SOCKET SEND FAILED!");
                perror("SOCKET SEND FAILED!");
                return -1;
            }
            return ret;//实际发送的数据长度
        }
        ssize_t NonBlockRecv(char* buff,ssize_t len){
            ssize_t ret = Recv(buff, len, MSG_DONTWAIT);
            return ret;//这个是正常返回的已经处理的ret
        }

        ssize_t NonBlockSend(const char*buff,ssize_t len){
            ssize_t ret = Send(buff, len, MSG_DONTWAIT);
            return ret;
        }

        //创建一个服务端连接
        bool CreateServer(uint16_t port = 8080,const std::string& ip="0.0.0.0",bool block_flag=false){
             //1. 创建套接字 2. 启动地址重用 3. 绑定地址，4. 开始监听，5. 设置非阻塞， 
            if(Create()==false)return false;

            if(block_flag==false)
                NonBLock();

            ReuseAddress();

            if (Bind(ip, 8080) == false)
                return false;
                
            if(Listen()==false)return false;
            
            return true;
        }

        //创建一个客户端连接
        bool CreateClient(uint16_t port = 8080,const std::string& ip="127.0.0.1"){
            if(Create()==false)
                return false;
            if(Connect(ip,port)==false)
                return false;
            return true;
        }

        //设置套接字为非阻塞属性
        void NonBLock(){
            //int fcntl(int fd, int cmd, ... /* arg */ );
            int flag = fcntl(_sockfd, F_GETFL,0);
            fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
        }

        //开启端口、ip重用
        void ReuseAddress(){
            // int setsockopt(int fd, int leve, int optname, void *val, int vallen)
            int val = 1;
            setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (const void *)&val, sizeof(val));
            val = 1;
            setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (const void *)&val, sizeof(val));
        }

        //关闭套接字//unistd.h
        void Close(){
            LOG(DBG, "close sockfd:%d", _sockfd);
            if(_sockfd!=-1){
                close(_sockfd);
                _sockfd = -1;
            }
        }
};

class Poller;
class EventLoop;
class Channel
{
    private:
        int _fd;//这个fd是套接字的fd
        EventLoop *_loop;
        uint32_t _events;
        uint32_t _revents;
        using EventCallback = std::function<void()>;
        EventCallback _read_callback;   //可读事件被触发的回调函数
        EventCallback _write_callback;  //可写事件被触发的回调函数
        EventCallback _error_callback;  //错误事件被触发的回调函数
        EventCallback _close_callback;  //连接断开事件被触发的回调函数
        EventCallback _event_callback;  //任意事件被触发的回调函数

    public:
        Channel(EventLoop *loop,int fd):_fd(fd),_loop(loop),_events(0),_revents(0){}
        int Fd(){ return _fd; }
        uint32_t Events() { return _events; }//获取想要监控的事件
        void SetEvents(uint32_t events) { _revents = events; }//设置就绪事件
        void SetReadCallback(const EventCallback &cb) { _read_callback = cb;Update();}
        void SetWriteCallback(const EventCallback &cb) { _write_callback = cb;Update();}
        void SetErrorCallback(const EventCallback &cb) { _error_callback = cb;Update(); }
        void SetCloseCallback(const EventCallback &cb) { _close_callback = cb;Update(); }
        void SetEventCallback(const EventCallback &cb) { _event_callback = cb;Update();}
        bool ReadAble() { return (_events & EPOLLIN); }
        bool WriteAble() { return (_events & EPOLLOUT); }
        void EnableRead() { _events |= EPOLLIN; Update();}
        void EnableWrite() { _events |= EPOLLOUT; Update();}
        void DisableRead(){_events &= ~EPOLLIN;Update();}
        void DisableWrite(){_events &= ~EPOLLOUT;Update();}
        void DisableAll(){_events = 0;Update();}
        void Remove();
        void Update(); // 每次修改events都要更新
        //这里的事件管理操作在下面实现，这里只有声明，因为需要Poller模块的实现
        // 事件处理，一旦连接触发了事件，就调用这个函数，自己触发了什么事件如何处理自己决定
        void HandleEvent(){
            //不管任何事件，都调用的回调函数

            if(_event_callback)_event_callback();
            if((_events&EPOLLIN)|(_events&EPOLLHUP)|(_events&EPOLLPRI)){
                //DBG_LOG("读事件处理");
                if (_read_callback)
                    _read_callback();
            }
            //有可能释放连接的事件只处理一次，避免重复释放
            if(_events&EPOLLOUT){
                //DBG_LOG("写事件处理");
                if (_write_callback)
                    _write_callback();
            }
            else if(_events&EPOLLERR){
                if(_error_callback)_error_callback();//一旦出错，就会释放连接，因此要放到前边调用任意回调
            }
            else if(_events&EPOLLHUP){
                if(_close_callback)_close_callback();
            }
        }
};
using TaskFunc  = std::function<void()>;//计数为0时执行计时任务
using ReleaseFunc = std::function<void()>;
class TimerTask
{
private:
    uint64_t _id;
    uint32_t _timeout;
    TaskFunc _task_cb;
    bool _canceled;
    ReleaseFunc _release;//用于删除TimerWheel中保存的定时器对象信息
public:
    TimerTask(uint64_t id,uint32_t timeout,TaskFunc task):_id(id),_timeout(timeout),_task_cb(task),_canceled(false)
    {}
    ~TimerTask(){
        if(_canceled==false)_task_cb();
        _release();
    }
    void setRelease(const ReleaseFunc&cb){
        _release = cb;
    }
    uint32_t delayTime(){
        return _timeout;
    }
    void Cancel(){
        _canceled = true;
    }
};

class TimerWheel{
    using PtrTask = std::shared_ptr<TimerTask>;
    using WeakTask = std::weak_ptr<TimerTask>;
    uint32_t _capacity;//容量，就是时间轮最大时长
    int _tick;//指针
    std::vector<std::vector<PtrTask>> _wheel;
    std::unordered_map<uint64_t, WeakTask> _timers;
    EventLoop *_loop;
    int _timerfd;//定时器描述符,可读事件就是读取定时器，执行定时任务
    std::unique_ptr<Channel> _timer_channel;

private:
    void RomoveTimer(uint64_t id){
        auto it = _timers.find(id);
        if(it!=_timers.end()){
            _timers.erase(it);
        }
    }
    
    static int CreateTimerfd(){
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if(timerfd < 0){
            DBG_LOG("TIMERFD CREATE FAILED!");
            return -1;
        }
        struct itimerspec tm;
        tm.it_interval.tv_nsec = 0;
        tm.it_interval.tv_sec = 1;
        tm.it_value.tv_nsec = 1;
        tm.it_value.tv_sec = 1;
        timerfd_settime(timerfd, 0, &tm, nullptr);
        return timerfd;
    }
    uint64_t ReadTimefd(){
        uint64_t times=0;
        //有可能因为其他描述符的事件处理花费事件比较长，然后在处理定时器描述符事件的时候，有可能就已经超时了很多次
        //read读取到的数据times就是从上一次read之后超时的次数
        int ret = read(_timerfd, &times, sizeof(times));
        if(ret < 0){
            ERR_LOG("READ TIMEFD FAILED!");
                abort();
        }
        return times;
    }
    // 这个函数应该每秒钟被执行一次，相当于秒针向后走了一步
    void RunTimerTask(){
        _tick = (_tick + 1)%_capacity;
        _wheel[_tick].clear();
    }
    void OnTime(){
        int times = ReadTimefd();//超时几次就向后移动几次
        for (int i = 0; i < times;i++){
            RunTimerTask();
        }
    }

    void TimerAddInLoop(uint64_t id,uint32_t timeout,const TaskFunc& task){
        //PtrTask p = std::make_shared<Timer_Task>(id, timeout, task);//使用参数构造
        PtrTask p = PtrTask(new TimerTask(id, timeout, task));//直接拷贝构造
        WeakTask p1 = WeakTask(p);//直接使用指针构造
        p->setRelease(std::bind(&TimerWheel::RomoveTimer, this, id));//将删除函数绑定给timertask
        int pos = (_tick + timeout) % _capacity;// timeout决定下标  //id是索引定时任务
        _timers[id] = p1;
        _wheel[pos].push_back(p);
    }
    void TimerRefreshInLoop(uint64_t id){
        auto it = _timers.find(id);
        if(it==_timers.end()){
            //没找到定时任务
            //不能增添新的，因为没有其他参数
            return;
        }
        PtrTask p = it->second.lock();//lock获取weak_ptr管理的对象对应的shared_ptr
        uint32_t delay = p->delayTime();//这个触发时间是固定时间
        int pos = (_tick + delay)%_capacity;//计算新的触发时间
        _wheel[pos].push_back(p);
    }
    void TimerCancelInLoop(uint64_t id){
        auto it = _timers.find(id);
        if(it==_timers.end()){
            return;
        }
        PtrTask p = it->second.lock();
        if(p)p->Cancel();
    }
public:
    TimerWheel(EventLoop* loop)
    :_capacity(60)
    , _tick(0)
    , _wheel(_capacity)
    ,_loop(loop)
    ,_timerfd(CreateTimerfd())
    ,_timer_channel(new Channel(_loop,_timerfd)) {
        _timer_channel->SetReadCallback(std::bind(&TimerWheel::OnTime, this));//设置读回调
        _timer_channel->EnableRead();//启动定时读监控
    }
    /*定时器中有个_timers成员，定时器信息的操作有可能在多线程中进行，因此需要考虑线程安全问题*/
        /*如果不想加锁，那就把对定期的所有操作，都放到一个线程中进行*/
    void TimerAdd(uint64_t id, uint32_t timeout, const TaskFunc &task);
    void TimerRefresh(uint64_t id);
    void TimerCancel(uint64_t id);
    /*这个接口存在线程安全问题--这个接口实际上不能被外界使用者调用，只能在模块内，在对应的EventLoop线程内执行*/
    bool HashTimer(uint64_t id){
        auto it = _timers.find(id);
        if(it != _timers.end())
            return true;
        return false;
    }
};

#define MAX_EPOLLEVENTS  1024
class Poller{
    private:
        int _epfd;//服务器只有一个epoll，一个epfd文件
        struct epoll_event _evs[MAX_EPOLLEVENTS];//
        std::unordered_map<int, Channel*> _channels;
    private:
        //对epoll直接操作
        void Update(Channel*channel,int op){
            // int epoll_ctl(int epfd, int op,  int fd,  struct epoll_event *ev);
            int fd = channel->Fd();
            struct epoll_event ev;
            ev.events = channel->Events();
            ev.data.fd = fd;
            assert(_epfd > 0);
            int ret = epoll_ctl(_epfd, op, fd, &ev);
            if(ret < 0){
                ERR_LOG("epoll_ctl error!");
                perror("epoll_ctl error!");
            }
            return;
        }
        //判断一个channel是否添加了事件监控
        bool HashChannel(Channel*channel){
            return _channels.count(channel->Fd());
        }

    public:
        Poller() { 
            _epfd = epoll_create(MAX_EPOLLEVENTS); 
            if(_epfd < 0){
                ERR_LOG("epoll_create error!");
                abort();//退出程序
            }
            //INF_LOG("epoll_create one");
        }
        // 添加或修改事件
        void UpdateEvent(Channel *channel)
        {
            bool ret = HashChannel(channel);
            if (ret)
            {
                Update(channel, EPOLL_CTL_MOD);
                return;
            }
            Update(channel, EPOLL_CTL_ADD);
            _channels.insert({channel->Fd(), channel});//
        }
        //删除事件
        void Remove(Channel* channel){
            bool ret = HashChannel(channel);
            if(ret){
                Update(channel, EPOLL_CTL_DEL);
                _channels.erase(channel->Fd());
            }
            return;
        }
        //开始监控，返回活跃连接
        void Epoll(std::vector<Channel*>*active){
            //为什么还在往上面调？
            //因为这是把触发事件交给eventloop处理
            // int epoll_wait(int epfd, struct epoll_event *evs, int maxevents, int timeout)
            //evs是触发事件
            int nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1);
            if (nfds < 0) {
                if (errno == EINTR) {
                    return ;
                }
                ERR_LOG("EPOLL WAIT ERROR:%s\n", strerror(errno));
                abort();//退出程序
            }
            for (int i = 0;i < nfds;i++){
                auto it = _channels.find(_evs[i].data.fd);
                assert(it != _channels.end());
                it->second->SetEvents(_evs[i].events);//给fd对应channel设置监控触发事件
                //这个fd对应channel指针是全局new的，在acceptor中创建
                //如果channel指针被删除了，就会段错误
                active->push_back(it->second);//往上面走，交付给eventloop
            }
        }
};



class EventLoop{
    private:
        std::thread::id _thread_id;
        Poller _poller;
        using Functor = std::function<void()>;
        std::vector<Functor> _tasks;//任务池
        int _event_fd;
        std::unique_ptr<Channel> _event_channel;//难道是eventfd的channel
        //eventfd 必须在eventchannel前面，不然无法初始化
        std::mutex _mutex;
        TimerWheel _timer_wheel;

    public:
        void RunInLoop(const Functor& ob)//判断任务是否是当前线程
        {
            if(IsInLoop())
                ob();
            else
                QueueInLoop(ob);
        }
        void QueueInLoop(const Functor& ob)//把当前任务添加到任务队列就绪
        //并且通过eventfd唤醒epoll
        {
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                _tasks.push_back(ob);
            }
            WeakUpEventFd();
        }
        bool IsInLoop()
        {
            return _thread_id == std::this_thread::get_id();
        }
        void AssertInLoop(){//注意thread是this_
            assert(_thread_id == std::this_thread::get_id());
        }
        //事件管理接口
        void UpdateEvent(Channel* channel)//添加、修改事件
        {
            _poller.UpdateEvent(channel);
        }
        void RemoveEvent(Channel* channel)//移除事件
        {
            _poller.Remove(channel);
        }

    public:
    //初始化
        EventLoop()
        :_thread_id(std::this_thread::get_id())
        ,_event_channel(new Channel(this,_event_fd))
        ,_event_fd(CreateEventFd())
        ,_timer_wheel(this)
        {
            _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventFd, this));
            //设置读事件回调，自动读取唤醒
            //接下俩启动读监控
            _event_channel->EnableRead();
        }
        void RunAllTask()//清空任务队列
        {
            std::vector<Functor> functor;
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                _tasks.swap(functor);
            }
            for(auto c:functor)
                c();
        }
        void Start(){
            while(1){
                //1.事件监控
                std::vector<Channel *> actives;
                _poller.Epoll(&actives);
                //DBG_LOG("触发事件，开始处理");

                //2.处理任意事件回调
                //在执行任务前执行回调
                for(auto c:actives){
                    c->HandleEvent();
                }

                //3.执行所有任务队列任务
                //不是上面的任务在这执行
                RunAllTask();
            }
        }

        //eventfd操作接口
        int CreateEventFd()
        {
            int initval = 0;
            int fd = eventfd(initval, EFD_CLOEXEC | EFD_NONBLOCK);
            if(fd < 0){
                perror("Eventfd Create Failed!");
                return -1;
            }
            return fd;
        }
        void WeakUpEventFd()
        {
            uint64_t initval = 1;
            int ret = write(_event_fd, &initval, sizeof(initval));
            if(ret < 0){
                //EINTR--被打断   EAGAIN--无数据可读
                if(errno == EINTR || errno ==EAGAIN){
                    return;
                }
                ERR_LOG("Wakeup Eventfd Failed!");
                abort();//退出程序，无法唤醒
            }
        }
        void ReadEventFd()
        {
            uint64_t res;
            int ret = read(_event_fd, &res, sizeof(res));
            if(ret < 0){
                if(errno == EINTR){
                    return;
                }
                ERR_LOG("Read Eventfd Failed!");
                abort();
            }
        }
        void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb) { return _timer_wheel.TimerAdd(id, delay, cb); }
        void TimerRefresh(uint64_t id) { return _timer_wheel.TimerRefresh(id); }
        void TimerCancel(uint64_t id) { return _timer_wheel.TimerCancel(id); }
        bool HashTimer(uint64_t id) { return _timer_wheel.HashTimer(id); }
};

void Channel::Update() { _loop->UpdateEvent(this); }
void Channel::Remove() { _loop->RemoveEvent(this); }
void TimerWheel::TimerAdd(uint64_t id, uint32_t timeout, const TaskFunc &task)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop, this, id, timeout, task));
}
void TimerWheel::TimerRefresh(uint64_t id){
    _loop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop, this, id));
}
void TimerWheel::TimerCancel(uint64_t id){
    _loop->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop, this, id));
}

class LoopThread{
    private:
        std::thread _thread;
        EventLoop *_loop;
        std::mutex _mutex;
        std::condition_variable _cond;
        void ThreadEntry() // 线程实体，即运行函数/任务
        {
            EventLoop loop;
            {
                std::unique_lock<std::mutex> lock(_mutex);//加锁
                _loop = &loop;
                _cond.notify_all();//唤醒
            }
            loop.Start();//eventloop实体只在这个作用域存在
        }

    public:
        LoopThread()//创建线程，设定入口函数
        :_loop(nullptr),_thread(std::thread(std::bind(&LoopThread::ThreadEntry,this))){}
        EventLoop* GetLoop(){
            EventLoop *loop = nullptr; // 为了避免线程安全问题，通过其他临时变量来获取变量
            // 而不直接返回
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _cond.wait(lock, [&]()
                           { return _loop != nullptr; });
                loop = _loop;
            }
            return loop;
        }
};
class LoopThreadPool{
    private:
        int _thread_count;
        int _next_idx;
        std::vector<LoopThread*> _threads;
        std::vector<EventLoop *> _loops;
        EventLoop *_baseloop;
    public:
        LoopThreadPool(EventLoop*base):_baseloop(base)//主线程已获取//从属线程还没有开启
        ,_thread_count(0)
        ,_next_idx(0){}
        void SetThreadCount(int count) { _thread_count = count; }
        
        //初始化
        void Create(){
            //创建线程池并获取eventloop
            if(_thread_count>0){
                _threads.resize(_thread_count);
                _loops.resize(_thread_count);
                for (int i = 0; i < _thread_count;i++){
                    _threads[i] = new LoopThread();
                    _loops[i] = _threads[i]->GetLoop();
                }
            }
        }

        //获取某个线程的eventloop  
        EventLoop* NextLoop(){
            if(_thread_count == 0)
                return _baseloop;
            _next_idx = (_next_idx + 1) % _thread_count;
            return _loops[_next_idx];
        }
};

class Any{
    private:
        class holder{
            public:
                virtual ~holder(){};
                //纯虚函数，子类必须重写
                virtual const std::type_info& type() = 0;
                virtual holder* clone() = 0;//子类重写提供拷贝的功能
                //这个拷贝是深拷贝
        };

        template<class T>
        class placeholder:public holder{
            public:
                placeholder(const T& a):_val(a){}
                virtual const std::type_info& type(){
                    return typeid(T);
                }
                virtual holder* clone(){//2025-4-28修改：如果和基类纯虚函数声明不一致就会成为抽象类，无法实例化
                    return new placeholder<T>(_val);
                }
                

            public:
                T _val;
        };
        holder* _content;
    public:
        Any():_content(nullptr){}
        template<class T>
        Any(const T& val):_content(new placeholder<T>(val)){}
        Any(const Any&other):_content(other._content?other._content->clone():nullptr){}
        ~Any() { delete _content; }
        template <class T>
        T *get() { return &((placeholder<T> *)_content)->_val; }
        Any& swap(Any &other) { //因为要交换指针，所以不能用const
            std::swap(other._content, _content);
            return *this;
        }
        Any& operator=(const Any &other) { 
            //为val构造一个临时的通用容器，然后与当前容器自身进行指针交换，临时对象释放的时候，原先保存的数据也就被释放
            Any temp(other);
            return swap(temp);
        }
        template<class T>
        Any& operator=(const T&val){
            Any temp(val);
            return swap(temp);
        }
};
class Connection;//因为定义指针类型需要在类作用域外面，所以要提前声明
enum ConnStatu
{
    DISCONNECTED,
    DISCONNECTING,
    CONNECTED,
    CONNECTING
};
using PtrConnection = std::shared_ptr<Connection>;
class Connection:public std::enable_shared_from_this<Connection>
{
private:
    uint64_t _conn_id; // 连接的唯一ID，便于连接的管理和查找
    EventLoop *_loop;   // 连接所关联的一个EventLoop
    int _sockfd;// 连接关联的文件描述符
    Socket _socket;// 套接字操作管理
    Channel _channel;// 连接的事件管理
    Buffer _in_buffer, _out_buffer;// 输入缓冲区---存放从socket中读取到的数据
    // 输出缓冲区---存放要发送给对端的数据
    Any _context; // 上下文
    ConnStatu _statu;//连接状态

    //超时销毁功能
    bool _enable_inactive_close;//非活跃关闭功能
    uint64_t _timer_id;//这里timerid使用conid，超时销毁和connection绑定

private:
    /*这四个回调函数，是让服务器模块来设置的（其实服务器模块的处理回调也是组件使用者设置的）*/
    /*换句话说，这几个回调都是组件使用者使用的*/
    //可以想象成java的切面日志
    using ConnectedCallback = std::function<void(const PtrConnection&)>;//连接真正创建回调
    using ClosedCallback = std::function<void(const PtrConnection&)>;//连接真正关闭回调
    using MessageCallback = std::function<void(const PtrConnection&,Buffer*buffer)>;//业务处理函数
    using AnyEventCallback = std::function<void(const PtrConnection&)>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;

    /*组件内的连接关闭回调--组件内设置的，因为服务器组件内会把所有的连接管理起来，一旦某个连接要关闭*/
    /*就应该从管理的地方移除掉自己的信息*/
    ClosedCallback _server_closed_callback;
public:
    int GetFd()//获取描述符，向外界提供
    {
        return _sockfd;
    }

private:
    //5个channel的回调函数
    //这几个是监控触发的回调
    //本来是应该在tcp_srv:main中配置的
    void HandleRead(){
        //1.接收socket的数据，放到缓冲区
        char buff[65536];
        int ret = _socket.NonBlockRecv(buff, 65535);
        if(ret < 0){
            //出错了，不能直接关闭连接
            //不直接关闭，半关闭
            return ShutdownInLoop();
        }
        _in_buffer.WriteAndPush(buff, ret);
        // 2.调用messgecallback业务处理
        if(_in_buffer.ReadAbleSize()>0){
            if(_message_callback)
                _message_callback(shared_from_this(), &_in_buffer);
        }
    }
    void HandleWrite(){
        //1.将缓冲区的数据传输到socket
        //写大小必须小于缓冲区可读大小
        //这点缓冲区提供了readablesize
        ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadPosition(), _out_buffer.ReadAbleSize());
        if(ret < 0){
            //发送错误就该关闭连接了,
            //但是要先处理残留数据
            if(_in_buffer.ReadAbleSize()>0)
                _message_callback(shared_from_this(),&_in_buffer);
            return Release();
        }

        _out_buffer.MoveReadOffset(ret);//因为是把传输多少数据交给socket，所以一定要记得手动moveoffset
        //2.如果缓冲区未0，且半关闭状态，要关闭连接
        if(_out_buffer.ReadAbleSize()==0)
        {
            _channel.DisableWrite();//没有写事件了，关闭写监控
            if (_statu == DISCONNECTING)
                return Release();
        }
    }
    void HandleClose()
    //描述符触发挂断事件
    {
        /*一旦连接挂断了，套接字就什么都干不了了，因此有数据待处理就处理一下，完毕关闭连接*/
        if (_in_buffer.ReadAbleSize() > 0)
        {
            _message_callback(shared_from_this(), &_in_buffer);
        }
        return Release();
    }
    void HandleError()
    //描述符触发出错事件
    {
        return HandleClose();
    }
    void HandleEvent()
    //描述符触发任意事件: 1. 刷新连接的活跃度--延迟定时销毁任务；  2. 调用组件使用者的任意事件回调
    {
        if(_enable_inactive_close==true)
            _loop->TimerRefresh(_timer_id);
        if(_event_callback)
            _event_callback(shared_from_this());
    }
private:
    void EstablishedInLoop()
    //连接创建成功后执行
    {
        //1.修改状态
        assert(_statu == CONNECTING);//当前的状态必须一定是上层的半连接状态
        _statu = CONNECTED;
        // 2.启动读监控
        _channel.EnableRead();
        // 3.调用服务器connected_callback回调
        if(_connected_callback)
            _connected_callback(shared_from_this());
    }
    void ReleaseInLoop(){
        //1.修改连接状态
        _statu = DISCONNECTED;
        // 2.移除连接的事件监控
        _channel.Remove();
        // 3.关闭描述符
        _socket.Close();
        // 4.取消定时器队列中的定时销毁任务
        if(_loop->HashTimer(_timer_id));
            CancelEnableInactiveRelease();
        // 5.调用服务器中的_server_closed_callback真正关闭回调函数
        if(_server_closed_callback)
            _server_closed_callback(shared_from_this());
    }
    void EnableInactiveReleaseInLoop(int timeout){
        //判断标志置为true
        _enable_inactive_close = true;
        //如果当前不存在定时任务则新增
        if(!_loop->HashTimer(_timer_id))
            _loop->TimerAdd(_timer_id, timeout, std::bind(&Connection::Shutdown, this));
        //存在则刷新
        _loop->TimerRefresh(_timer_id);
    }
    void CancelEnableInactiveReleaseInLoop(){
        //判断标志置为false
        _enable_inactive_close = false;
        //取消定时设置
        _loop->TimerCancel(_timer_id);
    }
    void SendInLoop(Buffer &data)
    //这个发送是写到缓冲区，不是直接发送
    //所以在释放连接处理残余数据的时候，发送数据不是把sendinloop(_out_buffer)
    //这样是把发送缓冲区发到发送缓冲区
    //自己写自己
    {
        if(_statu != DISCONNECTED)//可以为DISCONNECTING,因为可能在释放连接的过程中处理残留数据
            _out_buffer.WriteBufferAndPush(data);
        if (_channel.WriteAble() == false)
            _channel.EnableWrite();
    }
    void ShutdownInLoop(){
        //1.设置半关闭状态
        _statu = DISCONNECTING;
        // 2.读取剩余数据
        if(_in_buffer.ReadAbleSize()>0){
            if(_message_callback){
                _message_callback(shared_from_this(), &_in_buffer);
            }
        }
        // 3.如果写数据，就写（send同时会开启写监控）
        if(_out_buffer.ReadAbleSize()>0){
            if (_channel.WriteAble() == false)
                _channel.EnableWrite();
        }

        if(_out_buffer.ReadAbleSize()==0){
            ReleaseInLoop();
        }
    }
    void UpgradeInLoop(const ConnectedCallback&conn,const MessageCallback&msg,
        const AnyEventCallback&event,const ClosedCallback&close,const Any&context){
        _context = context;
        _connected_callback = conn;
        _message_callback = msg;
        _closed_callback = close;
        _event_callback = event;
    }

public:
    Connection(EventLoop*loop,uint64_t connid,int sockfd)
    :_loop(loop)
    ,_conn_id(connid)
    ,_sockfd(sockfd)
    ,_socket(_sockfd)
    ,_channel(_loop,_sockfd)
    ,_enable_inactive_close(false)
    ,_statu(CONNECTING)
    ,_timer_id(_conn_id)
    {
        _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
        _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
        _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
        _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
        _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
    }

    ~Connection(){
        DBG_LOG("RELEASE CONNECTION:%p", this);
    }
    // 获取管理的文件描述符
    int Fd() { return _sockfd; }
    // 获取连接ID
    int Id() { return _conn_id; }
    // 是否处于CONNECTED状态
    bool Connected() { return (_statu == CONNECTED); }
    // 设置上下文--连接建立完成时进行调用
    void SetMessageCallback(const MessageCallback &cb) { _message_callback = cb; }
    void SetConnectedCallback(const ConnectedCallback &cb) { _connected_callback = cb; }
    void SetClosedCallback(const ClosedCallback &cb) { _closed_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback &cb) { _event_callback = cb; }
    void SetSrvCloseCallback(const ClosedCallback &cb) { _server_closed_callback = cb; }
    void Send(const char*data,int len)//发送数据到cli
    {
        //Buffer buffer(data);
        //一个是修改构造函数
        //但是这样可能存在漏洞问题
        //一个是复用WriteAndPush给buffer
        Buffer buff;
        buff.WriteAndPush(data, len);
        return SendInLoop(buff);
    }
    void Shutdown()//不实际关闭
    {
        _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop, this));
    }
    void EnableInactiveRelease(int timeout)//启动连接超时销毁
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this,timeout));
    }
    void CancelEnableInactiveRelease()//关闭超时销毁
    {
        _loop->RunInLoop(std::bind(&Connection::CancelEnableInactiveReleaseInLoop, this));
    }
    void Upgrade(const ConnectedCallback&conn,const MessageCallback&msg,
        const AnyEventCallback&event,const ClosedCallback&close,const Any&context)//协议切换
    //重置上下文以及阶段性回调处理函数--需要在eventloop线程中立即执行
    //防备新的事件触发后，处理的时候，切换协议任务还没有被执行--导致数据使用原协议处理了
    {
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this,conn,msg,event,close,context));
    }
    void Established()//半连接状态
    {
        _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));
    }
    void Release()//彻底关闭释放连接
    //描述符套接字、事件监控、超时销毁、服务器信息
    {
        _loop->RunInLoop(std::bind(&Connection::ReleaseInLoop, this));
    }
    void SetContext(const Any &context) { _context = context; }
    Any*GetContext(){return &_context;}
};


class Acceptor
{
private:
    EventLoop *_loop;
    Socket _sock;
    Channel _channel;
    using AcceptCallBack = std::function<void(int)>;
    AcceptCallBack _accept_callback;//连接创建回调，这个是和epoll没有关系的回调
    //每次连接建立会附加执行
public:
    Acceptor(EventLoop *loop, int port)
        : _sock(CreateSocket(port)), _channel(loop, _sock.Fd()), _loop(loop)
    {
        _channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
    }
    void SetAcceptCallback(const AcceptCallBack &cb) { _accept_callback = cb; }
    void Listen() { _channel.EnableRead(); }

private:
    void HandleRead()
    {
        int newfd = _sock.Accept();
        if (newfd < 0)
        {
            return;
        }

        //进行连接处理
        if (_accept_callback)
            _accept_callback(newfd);
    }
    int CreateSocket(int port)
    {
        // 还可以先默认构造然后创建fd再返回？
        bool ret = _sock.CreateServer(port);
        assert(ret == true);
        return _sock.Fd();
    }


};


class TCP_Server{
    private:
        uint64_t _next_id;//这个id为什么是64位就是8字节？
        //好像这个id就是同时给conn和定时任务分配的
        uint16_t _port;
        bool _enable_inactive_release;
        int _timeout;
        EventLoop _baseloop;
        LoopThreadPool _pool; 
        Acceptor _acceptor;  
        std::unordered_map<uint64_t, PtrConnection> _conns; // 管理conn的shared_ptr//id是什么？


        using ConnectedCallback = std::function<void(const PtrConnection&)>;//连接真正创建回调
        using ClosedCallback = std::function<void(const PtrConnection&)>;//连接真正关闭回调
        using MessageCallback = std::function<void(const PtrConnection&,Buffer*buffer)>;//业务处理函数
        using AnyEventCallback = std::function<void(const PtrConnection&)>;
        using Functor = std::function<void()>;//虽然前面有定义，但是是私有的无法访问
        //这正是是因为类定义类型之间的解耦，避免名称相同功能不同混淆
        ConnectedCallback _connected_callback;
        MessageCallback _message_callback;
        ClosedCallback _closed_callback;
        AnyEventCallback _event_callback;
        /*组件内的连接关闭回调--组件内设置的，因为服务器组件内会把所有的连接管理起来，一旦某个连接要关闭*/
        /*就应该从管理的地方移除掉自己的信息*/
        ClosedCallback _server_closed_callback;

    public:
        TCP_Server(uint16_t port):_port(port)
        ,_enable_inactive_release(false),_acceptor(&_baseloop,_port)
        ,_pool(&_baseloop),_next_id(0)
        {
            _acceptor.SetAcceptCallback(std::bind(&TCP_Server::NewConnection,this,std::placeholders::_1));
            _acceptor.Listen();
        }

        //设置eventloop线程池数量
        void SetThreadCount(int threadcount){
            return _pool.SetThreadCount(threadcount);
        }
        void Start(){
            //acceptor设置连接建立回调
            //acceptor开始监听
            
            //pool初始化
            _pool.Create();

            //开始监控监听套接字事件
            _baseloop.Start();
        }
        void SetMessageCallback(const MessageCallback &cb) { _message_callback = cb; }
        void SetConnectedCallback(const ConnectedCallback &cb) { _connected_callback = cb; }
        void SetClosedCallback(const ClosedCallback &cb) { _closed_callback = cb; }
        void SetAnyEventCallback(const AnyEventCallback &cb) { _event_callback = cb; }
        void SetSrvCloseCallback(const ClosedCallback &cb) { _server_closed_callback = cb; }
    
        //开启连接超时销毁
        void EnableInactiveRelease(int timeout){
            _timeout = timeout;
            _enable_inactive_release = true;
        }

        //添加定时任务
        void RunAfter(const Functor& task,int delay){
            _baseloop.RunInLoop(std::bind(&TCP_Server::RunAfterInLoop, this, task, delay));
        }

    private:
        void RunAfterInLoop(const Functor& task,int delay){
            _next_id++;
            _baseloop.TimerAdd(_next_id, delay, task);
        }

        //accept_callback回调
        void NewConnection(int newfd){
            // 创建新连接connection
            _next_id++;
            PtrConnection conn(new Connection(_pool.NextLoop(), _next_id, newfd));
            //设置回调
            conn->SetConnectedCallback(_connected_callback);//连接建立回调
            conn->SetMessageCallback(_message_callback);//业务处理函数，外部传入
            conn->SetClosedCallback(_closed_callback);//连接开始关闭回调
            conn->SetAnyEventCallback(_event_callback);
            conn->SetSrvCloseCallback(std::bind(&TCP_Server::RemoveConnection, this, std::placeholders::_1));
            //真正关闭回调
            //开启超时连接销毁
            if(_enable_inactive_release==true)
                conn->EnableInactiveRelease(_timeout);
            _conns.insert({conn->Id(), conn});
            conn->Established();//就绪初始化
        }

        
        void RemoveConnectionInLoop(const PtrConnection& conn)
        //从tcp_server中删掉连接的管理，被srv_close回调
        {
            int id = conn->Id();
            auto it = _conns.find(id);
            if(it!=_conns.end()){
                _conns.erase(it);
            }
        }

        void RemoveConnection(const PtrConnection&conn){
            _baseloop.RunInLoop(std::bind(&TCP_Server::RemoveConnectionInLoop, this, conn));
        }
}; 

#endif